package com.xiaoke.mapreduce.topn;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;

public class TMapper extends Mapper<LongWritable, Text, TKey, IntWritable> {

    TKey mkey = new TKey();
    IntWritable mval = new IntWritable();
    public HashMap<String,String> dict = new HashMap<String,String>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
//        1    beijing
//        2    shanghai
//        3    Guangzhou
        URI[] cacheFiles = context.getCacheFiles();
        Path path = new Path(cacheFiles[0].getPath());
        BufferedReader bufferedReader= new BufferedReader(new FileReader(new File(path.getName())));
        String line = bufferedReader.readLine();
        while (line != null){
            String[] split = line.split(" ");
            dict.put(split[0], split[1]);
            line = bufferedReader.readLine();
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        //开发习惯：不要过于自信
        // value:  2019-6-1 22:22:22	1	31
        String[] split = StringUtils.split(value.toString(), '\t');
        // 2019-6-1 22:22:22
        // 1
        // 31

        try {
            Date data = sdf.parse(split[0]);
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(data);
            mkey.setYear(calendar.get(Calendar.YEAR));
            mkey.setMonth(calendar.get(Calendar.MONTH) + 1);
            mkey.setDay(calendar.get(Calendar.DAY_OF_MONTH));

            int wd = Integer.parseInt(split[2]);
            mkey.setWd(wd);
            mkey.setLocation(dict.get(split[1]));
            mval.set(wd);

            // TKey 是根据年月日排序
            // map这里将值拆成: 年月日温度  温度
            context.write(mkey, mval);
        } catch (ParseException e) {
            e.printStackTrace();
        }


    }
}
